﻿#include "TaskProcess.h"
#include <QMutexLocker>
#include <QEventLoop>
#include <QTimer>
#include "pipelinecheckdlg.h"
#include "componentmanager.h"
#include "systemconfiguration.h"
#include "../../pipeline/pipelinemanager.h"
#include "component_log.h"
#include "pathexcutequeue.h"
#include "taskinfosql.h"


void TaskProcess::Sleep(int msec)
{
	QEventLoop loop;
	QTimer::singleShot(msec, &loop, &QEventLoop::quit);
	loop.exec();
}

TaskProcess::TaskProcess(QObject* parent)
	: ComponentBase(parent)
{
	m_name = u8"任务处理组件";
	m_type = Component::component_taskExcutor;
}

TaskProcess::~TaskProcess()
{}

QList<Pipeline*> TaskProcess::GetPipeList()
{
	return m_pipelist;
}

PathExcuteQueue* TaskProcess::GetPathExcutor()
{
	return m_pathExcute;
}

bool TaskProcess::PipelineErrorClear(Pipeline* p)
{
	p->ClearErrorState();
	MovePipeFromWarningQueue(p);
	AddPipeToPrepareQueue(p);
	return true;
}

// 数据库存储任务处理，软件第一次启动时执行
void TaskProcess::SendSqlTask()
{
	if (m_sql == nullptr) {
		return;
	}
	QByteArrayList tasks;
	if (m_sql->ReadAllTask(tasks)) {
		for (auto& iter : tasks) {
			for (auto pipe : m_pipelist) {
				int ret = pipe->IfTaskAvailable(iter);
				if (ret == 0) {	// 任务匹配成功
					log_trace("{} recive task", pipe->GetPipelineId());
					QMutexLocker locker(&prepareLock);
					pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_receive);
					m_prepare.append(pipe);
					break;
				} else if (ret > 1) {	// 任务匹配成功，但是业务线繁忙
					SetWarningmsg(QString("{}收到任务，因为业务线存在正在执行中的任务而拒绝").arg(pipe->GetPipelineId()));
					break;
				} else {	// 任务未匹配成功
					//log_error("任务与{}不匹配", pipe->GetPipelineId());
					continue;
				}
			}
		}
	}
}

// 组件初始化处理函数
int TaskProcess::ComponentInit()
{
	// 订阅wcsmsg组件的任务消息
	auto wcs = m_ob->GetComponent(Component::component_wcsrec);
	if (wcs == nullptr) {
		SetErrmsg(u8"获取wcs消息组件未加载", false);
		return -99;
	}
	bool ret = wcs->RegisterSubscriberFunction((int)WCSMsgType::wcs_msg_task, [&](const QByteArray& task) {
		return this->OnDealWithTaskMsg(task);
		});
	if (!ret) {
		SetErrmsg(u8"wcs任务消息订阅失败", false);
		return -99;
	}

	auto platstack = m_ob->GetComponent(Component::component_plateStack);
	if (platstack != nullptr) {
		bool ret = platstack->RegisterSubscriberFunction(1, [&](const QByteArray& task) {
			return this->OnDealWithPlateTaskMsg(task);
			});
		if (!ret) {
			SetErrmsg(u8"托盘垛任务消息订阅失败", false);
			return -99;
		}
	}

	// 初始化pipeline
	auto pro = syscfg->GetProjectName();
	ret = PipelineManager::GetInstance()->LoadProjectPlugins(pro);
	if (!ret) {
		SetErrmsg(QString("加载项目%1插件失败").arg(pro), false);
		return -99;
	}
	QStringList ids = syscfg->GetPipelineids();
	for (auto& id : ids) {
		auto cfg = syscfg->GetPipelineCfgById(id);
		Pipeline* temp = PipelineManager::GetInstance()->CreatePipeline((PipelineType)cfg.type);
		if (temp != nullptr) {
			m_pipelist.append(temp);
			log_trace("pipeline [{}] load successed", id);
		} else {
			log_error("pipeline [{}] load failed", id);
			SetErrmsg("业务线加载失败", false);
			return -99;
		}
		if (!temp->PipelineInit(id)) {
			log_error("{} init failed", id);
			SetErrmsg(QString("业务线%1初始化失败").arg(id));
			return -99;
		}
	}

	m_pathExcute = new PathExcuteQueue(this);
	if (!m_pathExcute->Init()) {
		log_error("path excutor init failed");
		SetErrmsg("路径执行功能初始化失败", false);
		return -99;
	}
	connect(m_pathExcute, &PathExcuteQueue::PathExcutedFailed, this, &TaskProcess::OnPathExcutedFailed);
	connect(m_pathExcute, &PathExcuteQueue::PathParseFailed, this, &TaskProcess::OnPathExcutedFailed);

	m_sql = new TaskInfoSql();
	if (!m_sql->Init()) {
		SetWarningmsg("任务数据库启动失败，当前任务数据无法保存，软件关闭后会丢失");
		delete m_sql;
		m_sql = nullptr;
	} else {
		SendSqlTask();
	}

	m_taskThread = new std::thread(&TaskProcess::TaskRunning, this);
	m_stepThread = new std::thread(&TaskProcess::StepRunning, this);

	if (m_taskThread->joinable() && m_stepThread->joinable()) {
		m_taskThread->detach();
		m_stepThread->detach();
		log_trace("任务调度启动");
	} else {
		SetErrmsg("任务调度线程无法启动", false);
		return -99;
	}
	m_dlg = new PipelineCheckDlg();
	if (!m_dlg->Init(m_pipelist)) {
		SetErrmsg("业务线校验窗口初始化失败", false);
	}
	return 0;
}
/*自检接口*/
bool TaskProcess::ComponentCheckSelfState()
{

	return true;
}
/*复位接口*/
bool TaskProcess::ComponentSelfReset()
{
	m_dlg->StartReset();
	return true;
}

void TaskProcess::ResponseSystemState(int state)
{
	SystemStateType st = (SystemStateType)state;
	switch (st) {
	case SystemStateType::system_state_idle:
		TaskProcessStop();
		break;
	case SystemStateType::system_state_work:
		QTimer::singleShot(2000,this,&TaskProcess::TaskProcessStart);
		log_trace("task start for 2 s");
		break;
	case SystemStateType::system_state_stop:
		TaskProcessStop();
		break;
	default:
		log_error("system state error [{}]", state)
			break;
	}
}

bool TaskProcess::ExcuteCmd(int cmd, const QByteArray& input, QByteArray& output)
{
	if (cmd == 1) {	// 删除任务
		log_trace("Manual Delete task {}", input);
		return m_sql->DeleteTask(input);
	}
	if (cmd == 2) {	// 获取执行任务编号列表
		auto tasks = m_sql->GetTaskList();
		output = tasks.join(';').toUtf8();
		return true;
	}

	return false;
}


void TaskProcess::OnPathExcutedFailed(const QString& err)
{
	SetErrmsg(err);
}


void TaskProcess::SetErrmsg(const QString& str, bool send)
{
	TaskProcessStop();
	ComponentBase::SetErrmsg(str, send);
}

/*抓托盘消息处理*/
int TaskProcess::OnDealWithPlateTaskMsg(const QByteArray& data)
{
	for (auto pipe : m_pipelist) {
		int ret = pipe->IfTaskAvailable(data);
		if (ret == 0) {	// 任务匹配成功
			log_trace("{} recive plate task", pipe->GetPipelineId());
			pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_receive);
			QMutexLocker locker(&prepareLock);
			m_prepare.append(pipe);
			taskCond.notify_all();	// 通知任务线程，启动
			return 0;
		} else if (ret > 1) {	// 任务匹配成功，但是业务线繁忙
			SetWarningmsg(QString("{}收到任务，因为业务线存在正在执行中的任务而拒绝").arg(pipe->GetPipelineId()));
			return -1;
		} else {	// 任务未匹配成功
			//log_error("任务与{}不匹配", pipe->GetPipelineId());
			continue;
		}
	}
	log_error("task data not find pipe, data:{}", data);
	SetWarningmsg("接收到来自WCS的任务，但是任务未匹配的到对应业务线，请检查任务内容是否正确");
	return -1;
}

int TaskProcess::OnDealWithTaskMsg(const QByteArray& data)
{
	for (auto pipe : m_pipelist) {
		int ret = pipe->IfTaskAvailable(data);
		if (ret == 0) {	// 任务匹配成功
			log_trace("{} recive task", pipe->GetPipelineId());
			pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_receive);
			if (m_sql) {
				m_sql->InsertTask(data);
			}
			QMutexLocker locker(&prepareLock);
			m_prepare.append(pipe);
			taskCond.notify_all();	// 通知任务线程，启动
			return 0;
		} else if (ret > 1) {	// 任务匹配成功，但是业务线繁忙
			SetWarningmsg(QString("{}收到任务，因为业务线存在正在执行中的任务而拒绝").arg(pipe->GetPipelineId()));
			return -1;
		} else {	// 任务未匹配成功
			//log_error("任务与{}不匹配", pipe->GetPipelineId());
			continue;
		}
	}
	log_error("task data not find pipe, data:{}", data);
	SetWarningmsg("接收到来自WCS的任务，但是任务未匹配的到对应业务线，请检查任务内容是否正确");
	return -1;
}

bool TaskProcess::AddPipeToWarningQueue(Pipeline* p)
{
	QMutexLocker locker(&warningLock);
	for (auto& iter : m_warning) {
		if (iter == p) {
			log_error("pipe {} has been add to warning list", p->GetPipelineId());
			return false;
		}
	}
	m_warning.append(p);
	log_trace("pipe {} add to warning list successed", p->GetPipelineId());
	if (m_warning.size() == m_pipelist.size()) {
		SetErrmsg("所有业务线执行出错，等待恢复");
		m_stop = true;
	}

	return true;
}

bool TaskProcess::AddPipeToPrepareQueue(Pipeline* p)
{
	QMutexLocker locker(&prepareLock);
	for (auto& iter : m_prepare) {
		if (iter == p) {
			log_error("pipe {} has been add to m_prepare list", p->GetPipelineId());
			return false;
		}
	}
	m_prepare.append(p);
	taskCond.notify_all();
	log_trace("pipe {} add to m_prepare list successed", p->GetPipelineId());
	return true;
}

bool TaskProcess::AddPipeToExecQueue(Pipeline* p)
{
	QMutexLocker locker(&execLock);
	for (auto& iter : m_execqueue) {
		if (iter == p) {
			log_error("pipe {} has been add to m_execqueue list", p->GetPipelineId());
			return false;
		}
	}
	m_execqueue.append(p);
	stepCond.notify_all();
	log_trace("pipe {} add to m_execqueue list successed", p->GetPipelineId());
	return true;
}

bool TaskProcess::MovePipeFromWarningQueue(Pipeline* p)
{
	QMutexLocker locker(&warningLock);
	for (int i = 0; i < m_warning.size(); i++) {
		if (m_warning.at(i) == p) {
			m_warning.removeAt(i);
			log_trace("Move pipe {} from warning list successed", p->GetPipelineId());
			return true;
		}
	}
	return false;
}

bool TaskProcess::MovePipeFromExecQueue(Pipeline* p)
{
	QMutexLocker locker(&execLock);
	for (int i = 0; i < m_execqueue.size(); i++) {
		if (m_execqueue.at(i) == p) {
			m_execqueue.removeAt(i);
			log_trace("Move pipe {} from Execqueue list successed", p->GetPipelineId());
			return true;
		}
	}

	return false;
}

bool TaskProcess::MovePipeFromPrepareQueue(Pipeline* p)
{
	QMutexLocker locker(&prepareLock);
	for (int i = 0; i < m_prepare.size(); i++) {
		if (m_prepare.at(i) == p) {
			m_prepare.removeAt(i);
			log_trace("Move pipe {} from m_prepare list successed", p->GetPipelineId());
			return true;
		}
	}

	return false;
}


inline void Sleep(int ms)
{
	QEventLoop loop;
	QTimer::singleShot(ms, &loop, &QEventLoop::quit);
	loop.exec();
}
// 任务执行线程
void TaskProcess::TaskRunning()
{
	int index = 0;
	Pipeline* pipe = nullptr;
	log_trace("*************TaskRunning start****************************");
	while (1) {
		std::unique_lock<std::mutex> lock(taskLock);
		if (m_prepare.isEmpty()) {		// 读取准备列表，如果列表为空，当前线程暂停
			taskCond.wait(lock);
		} else {
			if (index >= m_prepare.size()) {
				index = 0;
			}
			pipe = m_prepare.at(index);	// 获取执行的业务线指针
		}

		if (m_stop || pipe == nullptr) {
			pipe = nullptr;
			index = 0;
			continue;
		}

		// 读取业务线状态根据状态，执行对应操作
		auto state = pipe->GetPipelineTaskState();
		switch (state) {
		case PipelineTaskState::pipeline_error:
			// 业务线出错的状态，将业务线添加到告警队列，并从准备队列中移除
			log_error("{} error", pipe->GetPipelineId());
			AddPipeToWarningQueue(pipe);
			MovePipeFromPrepareQueue(pipe);
			// 任务执行线程中，业务线出错，无需停止。
			SetWarningmsg(QString("因为%1，任务停止执行").arg(pipe->GetErrorMsg()));
			pipe = nullptr;
			index++;
			break;
		case PipelineTaskState::pipeline_task_idle:
			// 没有任务状态，直接从准备队列中移除
			log_error("{} is idle", pipe->GetPipelineId());
			MovePipeFromPrepareQueue(pipe);
			pipe = nullptr;
			index++;
			break;
		case PipelineTaskState::pipeline_task_receive:
			if (pipe->TaskInit()) {	// 任务初始化操作
				pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_inited);
			} else {
				// 任务初始化失败
				log_error("{} is task init failed", pipe->GetPipelineId());
				AddPipeToWarningQueue(pipe);
				MovePipeFromPrepareQueue(pipe);
				// 任务执行线程中，业务线出错，无需停止。
				SetWarningmsg(QString("因为%1，任务初始化操作执行失败").arg(pipe->GetErrorMsg()));
				pipe = nullptr;
				index++;
			}

			break;
		case PipelineTaskState::pipeline_task_inited:
			if (pipe->IfTaskStarted()) {
				pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_start);
			} else {	// 条件不足继续等待
				pipe = nullptr;
				index++;
			}
			break;
		case PipelineTaskState::pipeline_task_start:
			if (pipe->IfTaskPrepared()) {
				pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_wait_finish);
			} else {
				pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_stepPrepare);
			}

			break;
		case PipelineTaskState::pipeline_task_stepPrepare:
		{
			int ret = pipe->StepPrepare();
			if (ret == 0) {	// 节拍准备完成
				pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_stepStart);
			} else if (ret > 0) {	// 节拍准备条件不足
				pipe = nullptr;
				index++;
			} else {		// 节拍准备出错
				log_error("{} is step prepare failed", pipe->GetPipelineId());
				AddPipeToWarningQueue(pipe);
				MovePipeFromPrepareQueue(pipe);
				// 任务执行线程中，业务线出错，无需停止。
				SetWarningmsg(QString("因为%1，任务初始化操作执行失败").arg(pipe->GetErrorMsg()));
				pipe = nullptr;
				index++;
			}
		}
		break;
		case PipelineTaskState::pipeline_task_stepStart:
			if (pipe->IfStepStarted()) {
				// 节拍可以启动
				AddPipeToExecQueue(pipe);
				pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_start);
			}
			pipe = nullptr;
			index++;
			break;
		case PipelineTaskState::pipeline_task_wait_finish:
			if (pipe->IfTaskFinished()) {
				pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_finish);
			} else {
				pipe = nullptr;
				index++;
			}
			break;
		case PipelineTaskState::pipeline_task_finish:
			if (pipe->TaskFinish()) {
				pipe->SetPipelineTaskState(PipelineTaskState::pipeline_task_idle);
			} else {
				log_trace("{} is task finish", pipe->GetPipelineId());
				AddPipeToWarningQueue(pipe);

			}
			MovePipeFromPrepareQueue(pipe);
			index++;
			pipe = nullptr;
			break;
		default:
			break;
		}

	}
	log_error("*************TaskRunning Stop****************************");
}

// 节拍执行线程
void TaskProcess::StepRunning()
{
	log_trace("*************StepRunning start****************************");
	while (1) {
		std::unique_lock<std::mutex> lock(stepLock);
		if (m_execPipe == nullptr) {
			if (m_execqueue.isEmpty()) {		// 读取准备列表，如果列表为空，当前线程暂停
				stepCond.wait(lock);
			}
			m_execPipe = m_execqueue.takeFirst();
		}
		if (m_stop || m_execPipe == nullptr) {
			continue;
		}
		if (m_execPipe->IsError()) {
			m_stop = true;
			SetErrmsg(QString("节拍调度停止：%1").arg(m_execPipe->GetErrorMsg()));
			continue;
		}

		if (m_execPipe->IfStepPathEmpty()) {
			// 路径为空， 说明所有路径下发完成
			if (!m_execPipe->StepExecuteFinish()) {
				AddPipeToWarningQueue(m_execPipe);
				SetErrmsg(QString("节拍执行完成操作失败：%1").arg(m_execPipe->GetErrorMsg()));
			}
			m_execPipe = nullptr;
			continue;
		}
		if (!m_pathExcute->IfPathQueueIsIdle()) {
			Sleep(100);
			continue;
		}
		QString path;
		if (!m_execPipe->GetNextExecPath(path) || path.isEmpty()) {
			// 路径不为空，但是路径获取失败，说明路径还未生成，需要继续等待
			Sleep(100);
		} else {
			int ret = m_pathExcute->AddPathToQueue(m_execPipe, path);
			if (ret == -1) {
				AddPipeToWarningQueue(m_execPipe);
				m_execPipe = nullptr;
				SetErrmsg("节拍路径解析异常，无法执行");
			} else if (ret < -1) {
				AddPipeToWarningQueue(m_execPipe);
				m_execPipe = nullptr;
				SetErrmsg("添加到路径执行线程失败，程序bug");
			}
		}

	}
}



/*任务调度启动*/
void TaskProcess::TaskProcessStop()
{
	m_stop = true;
	if (m_pathExcute) {
		m_pathExcute->PathStop();
	}
	log_trace("TaskProcess stop");
}

void TaskProcess::TaskProcessStart()
{	
	m_stop = false;
	if (m_pathExcute) {
		m_pathExcute->PathStart();
	}
	log_trace("TaskProcess Start");
}
